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

package main

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"net/url"
	"regexp"
	"sort"
	"strings"
	"time"

	stdio "git.novox.be/novox/mesh-sdk/go"
)

// The ask the verb queues for the converter process (ADR 0468 §7; the shape of brother's device asks, ADR 0436 §6):
// on the work queue convert-asks, whose one consumer is the module's process `converter`, which writes its answer
// under the ask's id in the state convert-answers.
const (
	// EventAsked is the event on the work queue; the stream convert-asks carries it.
	EventAsked = "convert.asked"
	// AnswersState is the state the converter writes its answers in.
	AnswersState = "convert-answers"
	// ConvertWait is how long one call waits for the converter. The tool runner cuts every tool call at 30 s, so a
	// call answers within 25 s: a conversion that is not done by then goes on, and is read with convert {ask}.
	ConvertWait = 25 * time.Second
	// TakeUpWithin is how long an ask may wait on the queue before the converter takes it up: brother's device process
	// follows a conversion for 3 minutes.
	TakeUpWithin = 3 * time.Minute
)

// Types are the document types the converter takes (ADR 0468 §7). Any other is refused by name.
var Types = []string{"docx", "doc", "xlsx", "xls", "pptx", "ppt", "odt", "ods", "odp", "rtf", "txt", "html"}

// Ask is what the verb queues: the operation, when it was asked, and the operation's arguments.
type Ask struct {
	ID       string    `json:"id"`
	Op       string    `json:"op"`
	At       time.Time `json:"asked-at"`
	Deadline time.Time `json:"deadline"`
	// Source is the signed GET link of the document, Type its type, Target the signed PUT link of the PDF.
	Source string `json:"source"`
	Type   string `json:"type"`
	Target string `json:"target"`
	// SHA256 is the source's SHA-256 when the asker knows it; the converter then names the conversion by it.
	SHA256 string `json:"sha256,omitempty"`
}

var sha256Pattern = regexp.MustCompile(`^[0-9a-f]{64}$`)

// OpConvert is the one operation.
const OpConvert = "convert"

// AskBus is how the tool reaches the converter: it queues an ask and reads an answer back.
type AskBus interface {
	Publish(id string, body any) error
	Answer(id string) (json.RawMessage, error)
}

type sdkAskBus struct{}

func (sdkAskBus) Publish(id string, body any) error {
	return stdio.Emit(EventAsked, body, stdio.EmitOptions{MessageID: id})
}

func (sdkAskBus) Answer(id string) (json.RawMessage, error) {
	e, err := stdio.State(AnswersState).Get(id)
	if err != nil || e == nil {
		return nil, err
	}
	return e.Value, nil
}

// Service is the tool side.
type Service struct {
	Asks AskBus
	Now  func() time.Time
	// Wait is how long an answer is waited for, Poll how often it is looked for.
	Wait, Poll time.Duration
}

func newService() *Service {
	return &Service{Asks: sdkAskBus{}, Now: time.Now, Wait: ConvertWait, Poll: 500 * time.Millisecond}
}

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

// cleanType is a type as asked, lower-case and without a leading dot, when it is one the converter takes.
func cleanType(raw string) (string, error) {
	t := strings.ToLower(strings.TrimPrefix(strings.TrimSpace(raw), "."))
	for _, ok := range Types {
		if t == ok {
			return t, nil
		}
	}
	taken := append([]string(nil), Types...)
	sort.Strings(taken)
	return "", fmt.Errorf("type %q is not one the converter takes: %s", raw, strings.Join(taken, ", "))
}

func checkLink(name, raw string) error {
	u, err := url.Parse(strings.TrimSpace(raw))
	if err != nil || u.Host == "" || (u.Scheme != "http" && u.Scheme != "https") {
		return fmt.Errorf("%s is a signed http(s) link, such as one object-store.link gives", name)
	}
	return nil
}

// Convert queues one ask and waits for the converter's answer {size, sha256} (the PDF it put to target). Nothing is
// queued for a type the converter does not take or a link that is not one.
func (s *Service) Convert(ctx context.Context, source, typ, target, sha256 string) (map[string]any, error) {
	t, err := cleanType(typ)
	if err != nil {
		return nil, err
	}
	if err := checkLink("source", source); err != nil {
		return nil, err
	}
	if err := checkLink("target", target); err != nil {
		return nil, err
	}
	sha256 = strings.ToLower(strings.TrimSpace(sha256))
	if sha256 != "" && !sha256Pattern.MatchString(sha256) {
		return nil, fmt.Errorf("sha256 is the source's SHA-256, 64 hex digits, not %q", sha256)
	}
	at := s.Now().UTC()
	ask := Ask{ID: newAskID(), Op: OpConvert, At: at, Deadline: at.Add(TakeUpWithin), Source: strings.TrimSpace(source), Type: t, Target: strings.TrimSpace(target), SHA256: sha256}
	if err := s.Asks.Publish(ask.ID, ask); err != nil {
		return nil, fmt.Errorf("the ask for the converter could not be queued: %w", err)
	}
	return s.await(ctx, ask.ID)
}

// Read answers the ask of an earlier call: the converter's answer, or that it is still converting.
func (s *Service) Read(ctx context.Context, id string) (map[string]any, error) {
	if !askPattern.MatchString(id) {
		return nil, fmt.Errorf("ask %q is not an ask id: the 24 hex digits convert answered", id)
	}
	return s.await(ctx, id)
}

var askPattern = regexp.MustCompile(`^[0-9a-f]{24}$`)

// await waits at most Wait for the converter's answer under the id. A conversion still running is answered as
// converting with its ask id, never as a failure: it goes on, and convert {ask} reads its answer.
func (s *Service) await(ctx context.Context, id string) (map[string]any, error) {
	ctx, cancel := context.WithTimeout(ctx, s.Wait)
	defer cancel()
	for {
		raw, err := s.Asks.Answer(id)
		if err != nil {
			return nil, fmt.Errorf("the converter's answers cannot be read (ask %s): %w", id, err)
		}
		if raw != nil {
			var out map[string]any
			if err := json.Unmarshal(raw, &out); err != nil {
				return nil, fmt.Errorf("the converter's answer is not JSON (ask %s): %w", id, err)
			}
			if msg, bad := out["error"].(string); bad {
				return nil, fmt.Errorf("%s (ask %s)", msg, id)
			}
			out["ask"] = id
			return out, nil
		}
		select {
		case <-ctx.Done():
			return map[string]any{"state": "converting", "ask": id,
				"note": "the conversion goes on; call convert again with only ask to read its answer (kept 10 minutes)"}, nil
		case <-time.After(s.Poll):
		}
	}
}

func tools(s *Service) []stdio.Tool {
	return []stdio.Tool{{
		Name: "document-converter.convert",
		Description: "Turn a Word, spreadsheet, slide, text or web document into a PDF: the document is read from a signed GET " +
			"link and the PDF is put to a signed PUT link, so no document crosses the bus. Takes " + strings.Join(Types, ", ") +
			"; any other type is refused by name. Answers the PDF's size and SHA-256 or the document server's error; a conversion not done within 25 s is answered as converting with its ask id (a tool call is cut at 30 s), and convert with only ask reads it.",
		Input: map[string]any{
			"source": map[string]any{"type": "string", "description": "a signed GET link of the document"},
			"type":   map[string]any{"type": "string", "description": "the document's type: " + strings.Join(Types, ", ")},
			"target": map[string]any{"type": "string", "description": "a signed PUT link for the PDF"},
			"sha256": map[string]any{"type": "string", "description": "the document's SHA-256, if known (optional)"},
			"ask":    map[string]any{"type": "string", "description": "the ask id of a conversion that was answered as converting: alone, it reads that conversion's answer"},
		},
		Run: func(a map[string]any) (any, error) {
			str := func(k string) string { v, _ := a[k].(string); return v }
			if ask := str("ask"); ask != "" {
				if str("source") != "" || str("type") != "" || str("target") != "" || str("sha256") != "" {
					return nil, fmt.Errorf("ask is read on its own: give ask alone to read an earlier conversion, or source, type and target to start one")
				}
				return s.Read(context.Background(), ask)
			}
			return s.Convert(context.Background(), str("source"), str("type"), str("target"), str("sha256"))
		},
	}}
}