stemedb/sdk/go/steme/client.go

package steme

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"time"
)

// Client is the StemeDB HTTP API client with automatic signing.
//
// Create a client with NewClient(). All assertions are automatically
// signed using the provided Signer.
type Client struct {
	baseURL    string
	httpClient *http.Client
	signer     *Signer
}

// NewClient creates a new StemeDB API client.
//
// The baseURL should be the API endpoint (e.g., "http://localhost:3000").
// The signer is used to automatically sign all assertions.
//
// Example:
//
//	signer, _ := steme.GenerateSigner()
//	client := steme.NewClient("http://localhost:3000", signer)
func NewClient(baseURL string, signer *Signer) *Client {
	return &Client{
		baseURL: baseURL,
		httpClient: &http.Client{
			Timeout: 30 * time.Second,
		},
		signer: signer,
	}
}

// WithHTTPClient sets a custom http.Client.
//
// This is useful for adding custom timeouts, TLS config, or proxies.
func (c *Client) WithHTTPClient(httpClient *http.Client) *Client {
	c.httpClient = httpClient
	return c
}

// Assert creates a new assertion in the knowledge graph.
//
// The assertion is automatically signed using the client's signer.
// Returns the content-addressed hash of the created assertion.
//
// Example:
//
//	assertion := steme.NewAssertion("Tesla_Inc", "has_revenue").
//	    WithNumber(96.7).
//	    WithConfidence(0.95).
//	    WithSourceHash("0000000000000000000000000000000000000000000000000000000000000000").
//	    Build()
//
//	hash, err := client.Assert(ctx, assertion)
func (c *Client) Assert(ctx context.Context, assertion Assertion) (string, error) {
	// Validate before sending (includes source_hash requirement)
	if err := assertion.Validate(); err != nil {
		return "", err
	}

	// Sign the assertion
	signature, err := c.signAssertion(&assertion)
	if err != nil {
		return "", fmt.Errorf("steme: failed to sign assertion: %w", err)
	}

	assertion.Signatures = []SignatureEntry{signature}

	// Make the API request
	var resp CreateResponse
	if err := c.doJSON(ctx, "POST", "/v1/assert", assertion, &resp); err != nil {
		return "", err
	}

	return resp.Hash, nil
}

// Query queries assertions with optional filters and lens-based conflict resolution.
//
// Use QueryBuilder to construct query parameters fluently.
//
// Example:
//
//	params := steme.NewQuery().
//	    WithSubject("Tesla_Inc").
//	    WithPredicate("has_revenue").
//	    WithLens(steme.LensConsensus).
//	    Build()
//
//	result, err := client.Query(ctx, params)
func (c *Client) Query(ctx context.Context, params QueryParams) (*QueryResult, error) {
	queryURL := "/v1/query"

	// Build query parameters
	values := url.Values{}

	if params.Subject != nil {
		values.Add("subject", *params.Subject)
	}

	if params.Predicate != nil {
		values.Add("predicate", *params.Predicate)
	}

	if params.Lifecycle != nil {
		values.Add("lifecycle", string(*params.Lifecycle))
	}

	if params.Epoch != nil {
		values.Add("epoch", *params.Epoch)
	}

	if params.Lens != nil {
		values.Add("lens", string(*params.Lens))
	}

	if params.Limit > 0 {
		values.Add("limit", fmt.Sprintf("%d", params.Limit))
	}

	if len(values) > 0 {
		queryURL += "?" + values.Encode()
	}

	var result QueryResult
	if err := c.doJSON(ctx, "GET", queryURL, nil, &result); err != nil {
		return nil, err
	}

	return &result, nil
}

// Skeptic queries the "Trust but Verify" endpoint.
//
// Unlike Query which picks a winner via a lens, Skeptic shows all
// competing claims with their relative weights and conflict analysis.
//
// Example:
//
//	result, err := client.Skeptic(ctx, steme.SkepticQueryParams{
//	    Subject:   "Semaglutide",
//	    Predicate: "muscle_effect",
//	})
//
//	fmt.Printf("Status: %s, Conflict: %.2f\n", result.Status, result.ConflictScore)
//	for _, claim := range result.Claims {
//	    fmt.Printf("  %v (%.1f%% support)\n", claim.Value, claim.WeightShare*100)
//	}
func (c *Client) Skeptic(ctx context.Context, params SkepticQueryParams) (*SkepticResult, error) {
	queryURL := "/v1/skeptic"

	values := url.Values{}
	values.Add("subject", params.Subject)
	values.Add("predicate", params.Predicate)

	queryURL += "?" + values.Encode()

	var result SkepticResult
	if err := c.doJSON(ctx, "GET", queryURL, nil, &result); err != nil {
		return nil, err
	}

	return &result, nil
}

// Layered queries the per-source-class consensus endpoint.
//
// Returns tier-by-tier resolution showing how each source class
// (Regulatory, Clinical, Anecdotal, etc.) independently resolves
// the same claim, plus the overall winner from the highest-authority tier.
//
// Example:
//
//	result, err := client.Layered(ctx, steme.LayeredQueryParams{
//	    Subject:   "Semaglutide",
//	    Predicate: "muscle_effect",
//	})
//
//	for _, tier := range result.Tiers {
//	    fmt.Printf("Tier %d (%s): %v\n", tier.Tier, tier.SourceClass, tier.Winner)
//	}
func (c *Client) Layered(ctx context.Context, params LayeredQueryParams) (*LayeredResult, error) {
	queryURL := "/v1/layered"

	values := url.Values{}
	values.Add("subject", params.Subject)
	values.Add("predicate", params.Predicate)

	queryURL += "?" + values.Encode()

	var result LayeredResult
	if err := c.doJSON(ctx, "GET", queryURL, nil, &result); err != nil {
		return nil, err
	}

	return &result, nil
}

// Trace queries the audit log for agent decision tracing.
//
// Returns query audits for a specific agent within a time range, optionally
// filtered by subject pattern. This enables "Why did the agent think that?"
// debugging for incident investigation.
//
// Example:
//
//	result, err := client.Trace(ctx, steme.TraceParams{
//	    AgentID: "deadbeef...",
//	    From:    "1704067200",
//	    To:      "1704153600",
//	    Subject: "Tesla*",
//	    Limit:   50,
//	})
//
//	for _, audit := range result.Audits {
//	    fmt.Printf("Query %s at %d\n", audit.QueryID, audit.Timestamp)
//	}
func (c *Client) Trace(ctx context.Context, params TraceParams) (*TraceResult, error) {
	queryURL := "/v1/trace"

	// Build query parameters
	values := url.Values{}
	values.Add("agent_id", params.AgentID)
	values.Add("from", params.From)

	if params.To != "" {
		values.Add("to", params.To)
	}

	if params.Subject != "" {
		values.Add("subject", params.Subject)
	}

	if params.Limit > 0 {
		values.Add("limit", fmt.Sprintf("%d", params.Limit))
	}

	queryURL += "?" + values.Encode()

	var result TraceResult
	if err := c.doJSON(ctx, "GET", queryURL, nil, &result); err != nil {
		return nil, err
	}

	return &result, nil
}

// Supersede creates a supersession record for error correction.
//
// Supersession enables error correction without violating append-only semantics.
// Instead of mutating an assertion, we create a supersession record that points
// from the old (target) to the new (replacement) assertion.
//
// Example:
//
//	result, err := client.Supersede(ctx, steme.SupersedeParams{
//	    TargetHash:       "abc123...",
//	    SupersessionType: steme.SupersessionInvalidate,
//	    Reason:           "Proposal treated as approved. See incident INC-2024-001",
//	    NewHash:          "def456...",
//	    AgentID:          "deadbeef...",
//	    Signature:        "...",
//	})
func (c *Client) Supersede(ctx context.Context, params SupersedeParams) (*SupersedeResult, error) {
	var result SupersedeResult
	if err := c.doJSON(ctx, "POST", "/v1/supersede", params, &result); err != nil {
		return nil, err
	}

	return &result, nil
}

// Health checks the API health status.
//
// Returns the service status, version, and assertion count.
func (c *Client) Health(ctx context.Context) (*HealthResponse, error) {
	var resp HealthResponse
	if err := c.doJSON(ctx, "GET", "/v1/health", nil, &resp); err != nil {
		return nil, err
	}

	return &resp, nil
}

// HealthResponse represents the health check response.
type HealthResponse struct {
	Status          string `json:"status"`
	Version         string `json:"version"`
	AssertionsCount uint64 `json:"assertions_count"`
}

// signAssertion creates a signature for an assertion.
//
// The signature is over "{subject}:{predicate}" as raw bytes.
func (c *Client) signAssertion(a *Assertion) (SignatureEntry, error) {
	// Build canonical message for signing
	message, err := canonicalAssertionMessage(a)
	if err != nil {
		return SignatureEntry{}, err
	}

	// Create signature entry
	return c.signer.CreateSignature(message), nil
}

// canonicalAssertionMessage creates the canonical byte representation for signing.
//
// This must match the server's signature verification logic (worker.rs).
// Server expects: "{subject}:{predicate}" as raw bytes.
func canonicalAssertionMessage(a *Assertion) ([]byte, error) {
	return []byte(fmt.Sprintf("%s:%s", a.Subject, a.Predicate)), nil
}

// doJSON performs an HTTP request with JSON encoding/decoding.
func (c *Client) doJSON(ctx context.Context, method, path string, body any, result any) error {
	fullURL := c.baseURL + path

	var reqBody io.Reader
	if body != nil {
		jsonBytes, err := json.Marshal(body)
		if err != nil {
			return fmt.Errorf("steme: failed to marshal request: %w", err)
		}
		reqBody = bytes.NewReader(jsonBytes)
	}

	req, err := http.NewRequestWithContext(ctx, method, fullURL, reqBody)
	if err != nil {
		return fmt.Errorf("steme: failed to create request: %w", err)
	}

	if body != nil {
		req.Header.Set("Content-Type", "application/json")
	}

	// Add agent_id header for audit trail
	req.Header.Set("X-Agent-Id", c.signer.PublicKey())

	resp, err := c.httpClient.Do(req)
	if err != nil {
		return fmt.Errorf("steme: request failed: %w", err)
	}
	defer func() { _ = resp.Body.Close() }()

	// Read response body
	respBody, err := io.ReadAll(resp.Body)
	if err != nil {
		return fmt.Errorf("steme: failed to read response: %w", err)
	}

	// Check for HTTP errors
	if resp.StatusCode >= 400 {
		var apiErr struct {
			Error string `json:"error"`
			Code  string `json:"code"`
		}

		// Try to parse error response
		if err := json.Unmarshal(respBody, &apiErr); err == nil && apiErr.Error != "" {
			return &APIError{
				StatusCode: resp.StatusCode,
				Code:       apiErr.Code,
				Message:    apiErr.Error,
			}
		}

		// Fallback to generic error
		return &APIError{
			StatusCode: resp.StatusCode,
			Code:       "UNKNOWN",
			Message:    string(respBody),
		}
	}

	// Decode success response
	if result != nil {
		if err := json.Unmarshal(respBody, result); err != nil {
			return fmt.Errorf("steme: failed to decode response: %w", err)
		}
	}

	return nil
}

// QueryWithRetry polls until results are available or timeout.
//
// StemeDB is eventually consistent - assertions are durably written to the WAL
// but indexed asynchronously by the IngestWorker. Use this method when you need
// read-after-write consistency.
//
// For most agent workflows, prefer Query() directly and design for eventual
// consistency. This method is primarily useful for:
//   - Integration tests that need immediate verification
//   - User-facing applications requiring synchronous feedback
//   - Quick validation scripts
//
// The method polls every 50ms until TotalCount > 0 or maxWait is reached.
// Returns the last query result (which may have TotalCount=0 on timeout).
//
// Example:
//
//	result, err := client.QueryWithRetry(ctx, params, 5*time.Second)
//	if result.TotalCount == 0 {
//	    // Assertion not yet indexed - handle accordingly
//	}
func (c *Client) QueryWithRetry(ctx context.Context, params QueryParams, maxWait time.Duration) (*QueryResult, error) {
	deadline := time.Now().Add(maxWait)

	for {
		result, err := c.Query(ctx, params)
		if err != nil {
			return nil, err
		}
		if result.TotalCount > 0 {
			return result, nil
		}
		if time.Now().After(deadline) {
			// Return empty result, not error - timeout is expected behavior
			return result, nil
		}
		select {
		case <-ctx.Done():
			return nil, ctx.Err()
		case <-time.After(50 * time.Millisecond):
			// Continue polling
		}
	}
}