hyperguild/ingestion/internal/extract/docmark.go

// ingestion/internal/extract/docmark.go
package extract

import (
	"bytes"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"os"
	"time"
)

// docmarkRequest is a JSON-RPC 2.0 tools/call request for docmark's
// convert_to_markdown tool.
type docmarkRequest struct {
	JSONRPC string `json:"jsonrpc"`
	ID      int    `json:"id"`
	Method  string `json:"method"`
	Params  struct {
		Name      string `json:"name"`
		Arguments struct {
			ContentBase64 string `json:"content_base64"`
			Filename      string `json:"filename"`
		} `json:"arguments"`
	} `json:"params"`
}

type docmarkResponse struct {
	Result *struct {
		Content []struct {
			Type string `json:"type"`
			Text string `json:"text"`
		} `json:"content"`
		IsError bool `json:"isError"`
	} `json:"result"`
	Error *struct {
		Message string `json:"message"`
	} `json:"error"`
}

// extractViaDocmark converts path (PDF/DOCX/XLSX/PPTX/image) to Markdown by
// calling the docmark MCP server with a single self-contained tools/call
// request (docmark runs stateless_http -- no initialize handshake or session
// ID needed). DOCMARK_URL must be set (e.g.
// http://docmark.docmark.svc.cluster.local:3001/mcp); DOCMARK_BEARER_TOKEN
// is docmark's static bearer (network is docmark's primary auth boundary,
// this is defense-in-depth — ADR-0013).
func extractViaDocmark(path string) (string, error) {
	url := os.Getenv("DOCMARK_URL")
	if url == "" {
		return "", fmt.Errorf("extractViaDocmark: DOCMARK_URL is not set")
	}

	raw, err := os.ReadFile(path)
	if err != nil {
		return "", fmt.Errorf("read %s: %w", path, err)
	}

	var reqBody docmarkRequest
	reqBody.JSONRPC = "2.0"
	reqBody.ID = 1
	reqBody.Method = "tools/call"
	reqBody.Params.Name = "convert_to_markdown"
	reqBody.Params.Arguments.ContentBase64 = base64.StdEncoding.EncodeToString(raw)
	reqBody.Params.Arguments.Filename = fileBase(path)

	payload, err := json.Marshal(reqBody)
	if err != nil {
		return "", fmt.Errorf("marshal docmark request: %w", err)
	}

	httpReq, err := http.NewRequest(http.MethodPost, url, bytes.NewReader(payload))
	if err != nil {
		return "", fmt.Errorf("build docmark request: %w", err)
	}
	httpReq.Header.Set("Content-Type", "application/json")
	httpReq.Header.Set("Accept", "application/json, text/event-stream")
	if tok := os.Getenv("DOCMARK_BEARER_TOKEN"); tok != "" {
		httpReq.Header.Set("Authorization", "Bearer "+tok)
	}

	client := &http.Client{Timeout: 60 * time.Second}
	resp, err := client.Do(httpReq)
	if err != nil {
		return "", fmt.Errorf("call docmark: %w", err)
	}
	defer func() { _ = resp.Body.Close() }()

	body, err := io.ReadAll(resp.Body)
	if err != nil {
		return "", fmt.Errorf("read docmark response: %w", err)
	}

	if resp.StatusCode != http.StatusOK {
		return "", fmt.Errorf("docmark: HTTP %d: %s", resp.StatusCode, string(body))
	}

	jsonBody := body
	if data := sseDataPayload(body); data != nil {
		jsonBody = data
	}

	var out docmarkResponse
	if err := json.Unmarshal(jsonBody, &out); err != nil {
		return "", fmt.Errorf("decode docmark response: %w", err)
	}
	if out.Error != nil {
		return "", fmt.Errorf("docmark: %s", out.Error.Message)
	}
	if out.Result == nil || len(out.Result.Content) == 0 {
		return "", fmt.Errorf("docmark: empty response")
	}
	text := out.Result.Content[0].Text
	if out.Result.IsError {
		return "", fmt.Errorf("docmark: %s", text)
	}
	return text, nil
}

// sseDataPayload extracts the JSON payload from an SSE-framed response body
// ("event: message\r\ndata: {...}\r\n\r\n"). docmark's Streamable-HTTP
// transport frames every response this way (Content-Type: text/event-stream)
// regardless of stateless_http — that flag removes the session/initialize
// requirement, not the SSE wire framing. Returns nil if body isn't SSE-framed
// (e.g. a plain-JSON response, kept as a fallback for forward-compatibility).
func sseDataPayload(body []byte) []byte {
	const prefix = "data: "
	for _, line := range bytes.Split(body, []byte("\n")) {
		line = bytes.TrimRight(line, "\r")
		if bytes.HasPrefix(line, []byte(prefix)) {
			return bytes.TrimPrefix(line, []byte(prefix))
		}
	}
	return nil
}

// fileBase returns the final path segment (like filepath.Base, kept local to
// avoid importing path/filepath just for this one call).
func fileBase(path string) string {
	for i := len(path) - 1; i >= 0; i-- {
		if path[i] == '/' || path[i] == '\\' {
			return path[i+1:]
		}
	}
	return path
}