mesh-catalog/modules/only-office/cmd/only-office/converter.go

package main

// The converter process (novox/hq ADR 0468 §7, ADR 0436 §1, §3, §6; issue 1596). It runs under an account of its own
// and a bus account of its own, is the one consumer of the work queue convert-asks, serves no verb, and alone reads
// the document server's token (the JWT). For each ask it has the document server turn the source into a PDF, fetches
// the PDF from the document server and puts it to the ask's target link, and writes {size, sha256} -- or the error --
// under the ask's id in the state convert-answers, where the tool side reads it.

import (
	"bytes"
	"context"
	"crypto/hmac"
	"crypto/rand"
	"crypto/sha256"
	"encoding/base64"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"os"
	"os/signal"
	"strings"
	"syscall"
	"time"

	stdio "git.novox.be/novox/mesh-sdk/go"
	"git.novox.be/novox/mesh-sdk/go/objectlink"
	"git.novox.be/novox/mesh-sdk/go/process"
	"github.com/nats-io/nats.go/jetstream"
)

const (
	// AsksStream is the work-queue stream's bus name: the manifest's convert-asks, capitalised with its hyphen an underscore.
	AsksStream = "CONVERT_ASKS"
	// ConvertPath is the document server's conversion service, reached on the machine's own published port. The path is
	// the one the conversion API's request page names, https://api.onlyoffice.com/docs/docs-api/additional-api/conversion-api/request/
	// (read 2026-10-21): POST to https://documentserver/converter.
	ConvertPath = "/converter"
	// StaleAfter is how old an ask may be when the converter takes it: older than twice the longest wait, nobody waits.
	StaleAfter = 6 * time.Minute
	// ConvertRun is how long one conversion, its fetch and its put may take.
	ConvertRun = 4 * time.Minute
)

// errorMeanings are the conversion service's error codes. Read from the conversion API's documentation, page
// https://api.onlyoffice.com/docs/docs-api/additional-api/conversion-api/error-codes/ on 2026-10-21.
var errorMeanings = map[int]string{
	-1:  "unknown error",
	-2:  "conversion timeout error",
	-3:  "conversion error",
	-4:  "error while downloading the document file to be converted",
	-5:  "incorrect password",
	-6:  "error while accessing the conversion result database",
	-7:  "input error",
	-8:  "invalid token",
	-9:  "the converter cannot determine the output file format",
	-10: "size limit exceeded",
}

// meaningOf is an error code with its meaning, "-4: error while ...".
func meaningOf(code int) string {
	if m, ok := errorMeanings[code]; ok {
		return fmt.Sprintf("%d: %s", code, m)
	}
	return fmt.Sprintf("%d: a code the conversion API does not document", code)
}

// jwtHS256 signs a payload as a JSON Web Token with HMAC-SHA256.
func jwtHS256(secret string, payload []byte) string {
	enc := base64.RawURLEncoding
	signing := enc.EncodeToString([]byte(`{"alg":"HS256","typ":"JWT"}`)) + "." + enc.EncodeToString(payload)
	mac := hmac.New(sha256.New, []byte(secret))
	mac.Write([]byte(signing))
	return signing + "." + enc.EncodeToString(mac.Sum(nil))
}

type answerStore interface {
	Create(key string, value any) (uint64, error)
}

type action int

const (
	actionAck action = iota
	actionTerm
	actionRetry
)

// converterWorker answers asks. Everything it touches is handed in, so a test drives it with fakes.
type converterWorker struct {
	// JWT reads the document server's secret; only this process calls it.
	JWT func() (string, error)
	// Server is the document server's address, "http://127.0.0.1:9070".
	Server string
	// Docs talks to the document server, Store to the object store.
	Docs, Store *http.Client
	Answers     answerStore
	Now         func() time.Time
	Logf        func(string, ...any)
	NewKey      func() string
}

func (w *converterWorker) logf(f string, a ...any) {
	if w.Logf != nil {
		w.Logf(f, a...)
	}
}

// Handle takes one queued ask: it answers it under its id and says how the message is settled.
func (w *converterWorker) Handle(body []byte) (action, time.Duration) {
	var ask Ask
	if err := json.Unmarshal(body, &ask); err != nil || ask.ID == "" {
		w.logf("[only-office] converter: a queued ask could not be read: %v", err)
		return actionTerm, 0
	}
	if age := w.Now().Sub(ask.At); age > StaleAfter {
		w.logf("[only-office] converter: ask %s is %s old; nobody waits for it", ask.ID, age.Round(time.Second))
		return actionAck, 0
	}
	if !ask.Deadline.IsZero() && w.Now().After(ask.Deadline) {
		w.answer(ask.ID, map[string]any{"error": "the ask was taken up after its deadline: nothing was converted for it; ask again"})
		return actionTerm, 0
	}
	ctx, cancel := context.WithTimeout(context.Background(), ConvertRun)
	defer cancel()
	out, err := w.convert(ctx, ask)
	if err != nil {
		out = map[string]any{"error": err.Error()}
	}
	if err := w.answer(ask.ID, out); err != nil {
		return actionRetry, 5 * time.Second
	}
	return actionAck, 0
}

func (w *converterWorker) answer(id string, v map[string]any) error {
	if _, err := w.Answers.Create(id, v); err != nil && !errors.Is(err, stdio.ErrChanged) {
		w.logf("[only-office] converter: the answer to %s could not be written: %v", id, err)
		var api *jetstream.APIError
		if errors.As(err, &api) {
			return nil // a refused value would be refused again
		}
		return err
	}
	return nil
}

// convert does one ask: the document server converts, then the PDF is put to the target.
func (w *converterWorker) convert(ctx context.Context, ask Ask) (map[string]any, error) {
	if ask.Op != OpConvert {
		return nil, fmt.Errorf("the converter does not know the operation %q", ask.Op)
	}
	if _, err := cleanType(ask.Type); err != nil {
		return nil, err
	}
	secret, err := w.JWT()
	if err != nil {
		return nil, fmt.Errorf("the document server's token cannot be read: %w", err)
	}
	key := ask.SHA256
	if key == "" {
		key = w.NewKey()
	}
	req := map[string]any{"async": false, "filetype": ask.Type, "key": key, "outputtype": "pdf", "url": ask.Source}
	payload, _ := json.Marshal(req)
	req["token"] = jwtHS256(secret, payload) // JWT_IN_BODY: the document server reads the token from the body
	reqBody, _ := json.Marshal(req)
	hreq, err := http.NewRequestWithContext(ctx, http.MethodPost, w.Server+ConvertPath, bytes.NewReader(reqBody))
	if err != nil {
		return nil, err
	}
	hreq.Header.Set("Content-Type", "application/json")
	hreq.Header.Set("Accept", "application/json")
	resp, err := w.Docs.Do(hreq)
	if err != nil {
		return nil, fmt.Errorf("the document server cannot be reached: %w", err)
	}
	defer resp.Body.Close()
	raw, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
	if err != nil {
		return nil, fmt.Errorf("the document server's answer cannot be read: %w", err)
	}
	if resp.StatusCode != http.StatusOK {
		return nil, fmt.Errorf("the document server answered %d", resp.StatusCode)
	}
	var out struct {
		EndConvert bool   `json:"endConvert"`
		FileURL    string `json:"fileUrl"`
		Error      int    `json:"error"`
	}
	if err := json.Unmarshal(raw, &out); err != nil {
		return nil, fmt.Errorf("the document server's answer is not JSON: %w", err)
	}
	if out.Error != 0 {
		return nil, fmt.Errorf("the document server refused the conversion, error %s", meaningOf(out.Error))
	}
	if !out.EndConvert || out.FileURL == "" {
		return nil, errors.New("the document server did not finish the conversion: its answer has no file")
	}
	return w.deliver(ctx, out.FileURL, ask.Target)
}

// deliver fetches the PDF from the document server into a temporary file and puts it to the target.
func (w *converterWorker) deliver(ctx context.Context, fileURL, target string) (map[string]any, error) {
	greq, err := http.NewRequestWithContext(ctx, http.MethodGet, fileURL, nil)
	if err != nil {
		return nil, err
	}
	gresp, err := w.Docs.Do(greq)
	if err != nil {
		return nil, fmt.Errorf("the converted file cannot be fetched from the document server: %w", err)
	}
	defer gresp.Body.Close()
	if gresp.StatusCode != http.StatusOK {
		return nil, fmt.Errorf("the document server answered %d for the converted file", gresp.StatusCode)
	}
	tmp, err := os.CreateTemp("", "converted-*.pdf")
	if err != nil {
		return nil, fmt.Errorf("a temporary file cannot be made: %w", err)
	}
	defer os.Remove(tmp.Name())
	defer tmp.Close()
	n, err := io.Copy(tmp, io.LimitReader(gresp.Body, objectlink.MaxDocument+1))
	if err != nil {
		return nil, fmt.Errorf("the converted file cannot be read: %w", err)
	}
	if n > objectlink.MaxDocument {
		return nil, fmt.Errorf("the converted file is over %d bytes (novox/hq ADR 0468 §3)", objectlink.MaxDocument)
	}
	if _, err := tmp.Seek(0, io.SeekStart); err != nil {
		return nil, err
	}
	res, err := objectlink.Put(ctx, w.Store, target, tmp, n, objectlink.MaxDocument)
	if err != nil {
		return nil, fmt.Errorf("the store did not take the PDF: %w", err)
	}
	return map[string]any{"size": res.Size, "sha256": res.SHA256}, nil
}

func newKey() string {
	b := make([]byte, 16)
	rand.Read(b)
	return hex.EncodeToString(b)
}

func readSecret(path string) func() (string, error) {
	return func() (string, error) {
		raw, err := os.ReadFile(path)
		if err != nil {
			return "", err
		}
		s := strings.TrimSpace(string(raw))
		if s == "" {
			return "", fmt.Errorf("%s is empty", path)
		}
		return s, nil
	}
}

// converter is `only-office converter`: it reads its own bus account, connects, and consumes convert-asks until stopped.
func converter() int {
	brokerFile, jwtFile, server := os.Getenv("ONLY_OFFICE_BROKER"), os.Getenv("ONLY_OFFICE_JWT"), os.Getenv("ONLY_OFFICE_SERVER")
	if brokerFile == "" || jwtFile == "" || server == "" {
		fmt.Fprintln(os.Stderr, "[only-office] converter was started without ONLY_OFFICE_BROKER, ONLY_OFFICE_JWT or ONLY_OFFICE_SERVER")
		return 2
	}
	account, err := process.ReadAccount(brokerFile)
	if err != nil {
		fmt.Fprintf(os.Stderr, "[only-office] converter: %v\n", err)
		return 1
	}
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()
	w := &converterWorker{JWT: readSecret(jwtFile), Server: strings.TrimRight(server, "/"),
		Docs: &http.Client{Timeout: ConvertRun}, Store: &http.Client{Timeout: ConvertRun},
		Now: time.Now, NewKey: newKey, Logf: func(f string, a ...any) { fmt.Fprintf(os.Stderr, f+"\n", a...) }}
	fmt.Fprintf(os.Stderr, "[only-office] converter is consuming %s as %s\n", AsksStream, account.User)
	if err := runConverter(ctx, account, w); err != nil && ctx.Err() == nil {
		fmt.Fprintf(os.Stderr, "[only-office] converter stopped: %v\n", err)
		return 1
	}
	return 0
}

// runConverter connects as the converter's own account and consumes convert-asks until ctx ends, answering each ask in
// the state convert-answers. Apart from converter so a test can drive it over a bus of its own.
func runConverter(ctx context.Context, account process.Account, w *converterWorker) error {
	conn, err := process.Connect(account)
	if err != nil {
		return fmt.Errorf("the bus cannot be reached: %w", err)
	}
	defer conn.Close()
	w.Answers = conn.State(AnswersState)
	return conn.Consume(ctx, AsksStream, func(msg *process.Message) {
		act, delay := w.Handle(msg.Body)
		var err error
		switch act {
		case actionAck:
			err = msg.Ack()
		case actionRetry:
			err = msg.NakWithDelay(delay)
		case actionTerm:
			err = msg.Term()
		}
		if err != nil {
			w.logf("[only-office] converter: settling %s: %v", msg.Subject, err)
		}
	})
}