mesh-catalog/modules/brother/cmd/brother/printer.go

package main

import (
	"context"
	"crypto/rand"
	"encoding/base64"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"regexp"
	"slices"
	"strings"
	"time"

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

// pagePattern is a page of the admin site a caller may ask for: an absolute path under the site ending in
// .html, with no query, no dot-dot and no second slash at its start (which would name another host).
var pagePattern = regexp.MustCompile(`^/[A-Za-z0-9_./-]*\.html$`)

// DefaultPage is the page printer.settings reads when none is asked for.
const DefaultPage = "/home/status.html"

// The asks the verbs put to the device process (ADR 0468 clause 1; the shape of ADR 0436's mail-sends): the tools
// never hold a credential, so each verb queues an ask on the work-queue stream device-asks, and the device process,
// an account of its own, answers under the ask's id in the state device-answers.
const (
	// EventAsked is the event on the work queue; the stream device-asks carries it.
	EventAsked = "device.asked"
	// AsksStream is the work-queue stream's name on the bus: the manifest's local name device-asks, capitalised
	// with its hyphen an underscore, as the controller composes a stream's own bus name.
	AsksStream = "DEVICE_ASKS"
	// AnswersState is the state the device process writes its answers in.
	AnswersState = "device-answers"
	// DeviceProcessID is the device process's id as module.json's resource "device" names it: the controller
	// delivers its bus account as the own secret broker-device.
	DeviceProcessID = "device"
	// PrintTakeUp is how long a print ask may wait on the queue before the device process takes it up, and PrintFollow
	// how long the tool side follows it to its end: the device process may fetch and convert before the job. A call
	// answers within QuickWait whatever these are, and the print goes on (the tool runner cuts every call at 30 s).
	PrintTakeUp = 3 * time.Minute
	PrintFollow = 6 * time.Minute
	// AnswerWait is how long a verb waits for the device process; ScanWait how long a scan does.
	AnswerWait = 60 * time.Second
	ScanWait   = 180 * time.Second
	// QuickWait is how long a feeder scan is waited for in one call: the mesh console gives up on a call after
	// 30 seconds, so a longer scan is answered as started, with its id, and read later (issue 1552).
	QuickWait = 25 * time.Second
	// FeederWait is how long the tool waits in the background for a feeder scan: the longest the device process
	// lets one run (FeederMaxBound) and two minutes to store it.
	FeederWait = FeederMaxBound + 2*time.Minute
	// StaleAfter is how old an ask may be when the device process takes it: older than twice the longest wait,
	// nobody waits for it any more, so the device is not touched for it.
	StaleAfter = 2 * ScanWait
)

// The operations of an ask.
const (
	OpSettings = "settings"
	OpScan     = "scan"
	OpDocument = "document"
	OpUpload   = "upload"
	OpPrint    = "print"
	// OpOpen asks which documents of several pages are open in the bucket (issue 1538).
	OpOpen = "open-documents"
)

// Ask is what a verb queues: the operation, when it was asked, and the operation's own arguments.
type Ask struct {
	ID string    `json:"id"`
	Op string    `json:"op"`
	At time.Time `json:"asked-at"`
	// Deadline is when the asker stops waiting for the device process to take the ask up: one taken after it is
	// dropped before any work, so nothing is scanned or printed for an asker that has gone (issue 1539).
	Deadline time.Time `json:"deadline,omitempty"`

	Page   string `json:"page,omitempty"`   // settings
	Links  bool   `json:"links,omitempty"`  // settings: the page's links are answered too
	Hidden bool   `json:"hidden,omitempty"` // settings: the hidden fields are answered too
	Object string `json:"object,omitempty"` // document, print: a scan's key
	Upload string `json:"upload,omitempty"` // upload, print: an upload's id
	Part   int    `json:"part,omitempty"`   // document, upload: counted from 1
	Parts  int    `json:"parts,omitempty"`  // upload
	SHA256 string `json:"sha256,omitempty"` // upload: the whole document's
	Data   string `json:"data,omitempty"`   // upload: the part, base64
	Copies int    `json:"copies,omitempty"` // print
	Sides  string `json:"sides,omitempty"`  // print
	Media  string `json:"media,omitempty"`  // print
	Source string `json:"source,omitempty"` // scan: glass (default) or feeder
	Duplex bool   `json:"duplex,omitempty"` // scan: both sides of each sheet, feeder only
	Apart  bool   `json:"apart,omitempty"`  // scan: glass only, each of several documents lying apart is its own file

	Document string `json:"document,omitempty"` // scan: start, or an open document's id: a glass page into it
	Finish   bool   `json:"finish,omitempty"`   // scan: close the document named in Document into one PDF

	// Path with Node are what printer.print is asked for a file of a machine: the tool side turns them into an Object
	// before anything is queued, so the queued ask carries a key and the options and never these (ADR 0468 clause 5).
	// A URL is queued as given: the device process alone fetches it (clause 6).
	Path string `json:"-"`             // print: a file on the machine Node
	Of   string `json:"-"`             // print: the ask id of an earlier print, whose outcome is read
	Node string `json:"-"`             // print: the machine Path is on
	URL  string `json:"url,omitempty"` // print: a web document the device process fetches (issue 1599)
}

// AskBus is how the tools reach the device process: they queue an ask and read 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
}

// answerWait, scanWait and answerPoll are what the tools wait with; a test shortens them.
var (
	answerWait = AnswerWait
	scanWait   = ScanWait
	quickWait  = QuickWait
	feederWait = FeederWait
	// scanRun and feederRun are how long the device process may take over a scan once it has taken it up (its own
	// bounds, deviceproc.go); the tool follows a scan for its start-by wait and this.
	scanRun    = 5 * time.Minute
	feederRun  = FeederMaxBound + 5*time.Minute
	answerPoll = 500 * time.Millisecond
)

// askBus is the bus the tools use; a test puts a bus of its own in its place.
var askBus AskBus = sdkAskBus{}

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

// call queues the ask and waits for the device process's answer, at most wait. An answer that is an error is
// the error. Nothing here reads, opens or sends a credential (ADR 0436 §12).
func (s *Service) call(ctx context.Context, ask Ask, wait time.Duration) (map[string]any, error) {
	id, err := s.queue(ask, wait)
	if err != nil {
		return nil, err
	}
	return s.await(ctx, id, wait)
}

// queue puts the ask on the work queue and answers its id, under which the answer is written. The asker waits
// startBy for the device process to take the ask up (0 is for as long as it is waited for).
func (s *Service) queue(ask Ask, startBy time.Duration) (string, error) {
	if ask.ID == "" {
		ask.ID = newAskID()
	}
	ask.At = s.Now().UTC()
	if startBy > 0 {
		ask.Deadline = ask.At.Add(startBy)
	}
	if err := s.Asks.Publish(ask.ID, ask); err != nil {
		return "", fmt.Errorf("the ask for the device process could not be queued: %w", err)
	}
	return ask.ID, nil
}

// await waits at most wait for the answer to the ask with this id. An answer that is an error is the error.
func (s *Service) await(ctx context.Context, id string, wait time.Duration) (map[string]any, error) {
	poll := s.AnswerPoll
	if wait <= 0 {
		wait = AnswerWait
	}
	if poll <= 0 {
		poll = 500 * time.Millisecond
	}
	ctx, cancel := context.WithTimeout(ctx, wait)
	defer cancel()
	for {
		raw, err := s.Asks.Answer(id)
		if err != nil {
			return nil, fmt.Errorf("the device process's answers cannot be read: %w", err)
		}
		if raw != nil {
			var out map[string]any
			if err := json.Unmarshal(raw, &out); err != nil {
				return nil, fmt.Errorf("the device process's answer is not JSON: %w", err)
			}
			if msg, bad := out["error"].(string); bad {
				return nil, fmt.Errorf("%s", msg)
			}
			return out, nil
		}
		select {
		case <-ctx.Done():
			return nil, fmt.Errorf("the device process did not answer within %s: it may not be running, or the device is slow (ask %s)",
				wait, id)
		case <-time.After(poll):
		}
	}
}

// PrinterSettings asks the device process to log in, read one page of the admin site and log out, and waits for
// its answer. Nothing is queued before the page is checked and printer and printer-certificate are known to be set.
func (s *Service) PrinterSettings(ctx context.Context, page string, links, hidden bool) (map[string]any, error) {
	if page == "" {
		page = DefaultPage
	}
	if !validPage(page) {
		return nil, fmt.Errorf("page %q is not a page of the admin site: it starts with / and ends in .html", page)
	}
	if err := s.Settings.MissingPrinter(); err != nil {
		return nil, err
	}
	return s.call(ctx, Ask{Op: OpSettings, Page: page, Links: links, Hidden: hidden}, s.AnswerWait)
}

func validPage(page string) bool {
	return pagePattern.MatchString(page) && !strings.Contains(page, "..") && !strings.Contains(page, "//")
}

var (
	scanKeyPattern = regexp.MustCompile(`^[0-9a-f]{64}\.pdf$`)
	// printKeyPattern is a document the bucket keeps under its SHA-256: a scan's key, or the key node-files.upload
	// answers for a file sent from a machine (the bare SHA-256).
	printKeyPattern = regexp.MustCompile(`^[0-9a-f]{64}(\.pdf)?$`)
	uploadPattern   = regexp.MustCompile(`^[A-Za-z0-9_-]{1,64}$`)
	mediaPattern    = regexp.MustCompile(`^[a-z0-9_.-]{1,64}$`)
	sha256Pattern   = regexp.MustCompile(`^[0-9a-f]{64}$`)
)

// Sides are the values the printer's sides take.
var sidesValues = []string{"one-sided", "two-sided-long-edge", "two-sided-short-edge"}

// DefaultMedia is the paper printer.print takes when none is named.
const DefaultMedia = "iso_a4_210x297mm"

// checkDocument is what is wrong with an ask for a scan's part, or nil.
func checkDocument(a Ask) error {
	if !scanKeyPattern.MatchString(a.Object) {
		return fmt.Errorf("object %q is not a scan's key: 64 hex digits and .pdf", a.Object)
	}
	if a.Part < 1 {
		return fmt.Errorf("part is a whole number from 1")
	}
	return nil
}

// checkUpload is what is wrong with an ask for an upload's part, or nil: the numbers, the SHA-256, the size.
func checkUpload(a Ask) error {
	if !uploadPattern.MatchString(a.Upload) {
		return fmt.Errorf("upload %q is not an upload id: letters, digits, - and _, at most 64", a.Upload)
	}
	if a.Parts < 1 || a.Parts > MostParts {
		return fmt.Errorf("parts is a whole number from 1 to %d, not %d", MostParts, a.Parts)
	}
	if a.Part < 1 || a.Part > a.Parts {
		return fmt.Errorf("part %d is out of range: the upload has parts 1 to %d", a.Part, a.Parts)
	}
	if !sha256Pattern.MatchString(a.SHA256) {
		return fmt.Errorf("sha256 is the whole document's SHA-256, 64 lower-case hex digits")
	}
	raw, err := base64.StdEncoding.DecodeString(a.Data)
	if err != nil {
		return fmt.Errorf("data is not base64: %v", err)
	}
	if len(raw) > MaxPart {
		return fmt.Errorf("a part is at most %d bytes (700 KiB), not %d", MaxPart, len(raw))
	}
	return nil
}

// checkPrint is what is wrong with an ask to print, or nil, after the defaults are laid in.
func checkPrint(a *Ask) error {
	if a.Upload != "" {
		return fmt.Errorf("upload is no longer a source of printer.print: send the file with path and node, or name its key in object")
	}
	if (a.Path == "") != (a.Node == "") {
		return fmt.Errorf("path and node go together: path is a file on the machine node names")
	}
	sources := 0
	for _, given := range []bool{a.Object != "", a.Path != "", a.URL != ""} {
		if given {
			sources++
		}
	}
	if sources != 1 {
		return fmt.Errorf("name exactly one of object (a key in the bucket), path with node (a file on that machine) or url, not %s", map[bool]string{true: "none", false: "two or more"}[sources == 0])
	}
	if a.Object != "" && !printKeyPattern.MatchString(a.Object) {
		return fmt.Errorf("object %q is not a stored document's key: 64 hex digits, with .pdf for a scan", a.Object)
	}
	if a.Copies == 0 {
		a.Copies = 1
	}
	if a.Copies < 1 || a.Copies > 99 {
		return fmt.Errorf("copies is a whole number from 1 to 99, not %d", a.Copies)
	}
	if a.Sides == "" {
		a.Sides = "one-sided"
	}
	if !slices.Contains(sidesValues, a.Sides) {
		return fmt.Errorf("sides is one of %s, not %q", strings.Join(sidesValues, ", "), a.Sides)
	}
	if a.Media == "" {
		a.Media = DefaultMedia
	}
	if !mediaPattern.MatchString(a.Media) {
		return fmt.Errorf("media %q is not a media size name such as %s", a.Media, DefaultMedia)
	}
	return nil
}

// ScanDocument asks for one part of a scan kept in the bucket.
func (s *Service) ScanDocument(ctx context.Context, object string, part int) (map[string]any, error) {
	a := Ask{Op: OpDocument, Object: object, Part: part}
	if err := checkDocument(a); err != nil {
		return nil, err
	}
	return s.call(ctx, a, s.AnswerWait)
}

// PrinterUpload sends one part of a document to be printed later, and says whether the upload is complete.
func (s *Service) PrinterUpload(ctx context.Context, a Ask) (map[string]any, error) {
	a.Op = OpUpload
	if err := checkUpload(a); err != nil {
		return nil, err
	}
	return s.call(ctx, a, s.AnswerWait)
}

// sdkUpload asks node's node-files to send a file of that machine into the bucket and answers the key it was
// stored under (issue 1597). The form is the seat's verb on one machine, `seat:<seat>.<verb>@<node>`.
func sdkUpload(node, path, bucket string) (string, error) {
	raw, err := stdio.Ask("seat:node-files.upload@"+node, map[string]any{"path": path, "bucket": bucket})
	if err != nil {
		return "", err
	}
	var ans struct {
		Key    string          `json:"key"`
		Answer json.RawMessage `json:"answer"`
	}
	if err := json.Unmarshal(raw, &ans); err != nil {
		return "", fmt.Errorf("node-files' answer is not JSON: %w", err)
	}
	if ans.Key == "" && len(ans.Answer) > 0 {
		_ = json.Unmarshal(ans.Answer, &ans)
	}
	if ans.Key == "" {
		return "", fmt.Errorf("node-files' answer names no key")
	}
	return ans.Key, nil
}

// PrinterPrint prints a stored document, a file of a machine, or a web link, in one call. A file of a machine is first
// sent into the bucket by that machine's node-files (refused there by name when it is a secret, and then nothing is
// printed); the print ask the device process takes carries the key and the options and no bytes.
//
// The tool runner cuts every call at 30 s, so the call answers within QuickWait, upload included. A print not done by
// then goes on in the background and is answered as started, with its ask id; printer.print {ask} reads its outcome.
func (s *Service) PrinterPrint(ctx context.Context, a Ask) (map[string]any, error) {
	if a.Of != "" {
		if a.Object != "" || a.Path != "" || a.Node != "" || a.URL != "" || a.Upload != "" {
			return nil, fmt.Errorf("ask is read on its own: name exactly one of object, path with node, url, or the ask id of an earlier print")
		}
		return s.PrintResult(ctx, a.Of)
	}
	a.Op = OpPrint
	if err := checkPrint(&a); err != nil {
		return nil, err
	}
	if s.Settings.Printer == "" {
		return nil, unsetError("printer")
	}
	bucket := ""
	switch {
	case a.URL != "":
		if err := CheckURLScheme(a.URL); err != nil {
			return nil, err
		}
	case a.Path != "":
		if s.Upload == nil {
			return nil, fmt.Errorf("this module has no way to ask node-files for an upload")
		}
		var err error
		if bucket, err = s.Bucket(); err != nil {
			return nil, err
		}
	}
	id := newAskID()
	j := s.Jobs.put(id)
	go func() {
		j.body, j.err = s.printRun(id, a, bucket)
		close(j.done)
	}()
	return s.printWait(ctx, id, j)
}

// printRun is a print followed to its end, however long the caller waits: the upload, the ask, the device's answer.
func (s *Service) printRun(id string, a Ask, bucket string) (map[string]any, error) {
	from := map[string]any{}
	if a.Path != "" {
		key, err := s.Upload(a.Node, a.Path, bucket)
		if err != nil {
			return nil, fmt.Errorf("%s on %s could not be sent to the bucket, and nothing is printed: %w", a.Path, a.Node, err)
		}
		from["path"], from["node"] = a.Path, a.Node
		a.Object, a.Path, a.Node = key, "", ""
	}
	a.ID = id
	if _, err := s.queue(a, PrintTakeUp); err != nil {
		return nil, err
	}
	ctx, cancel := context.WithTimeout(context.Background(), PrintFollow)
	defer cancel()
	out, err := s.await(ctx, id, PrintFollow)
	if err != nil {
		return nil, err
	}
	for k, v := range from {
		out[k] = v
	}
	return out, nil
}

// printWait gives the print's outcome if it comes within QuickWait, and started with the ask id otherwise.
func (s *Service) printWait(ctx context.Context, id string, j *job) (map[string]any, error) {
	quick := s.QuickWait
	if quick <= 0 {
		quick = QuickWait
	}
	select {
	case <-j.done:
		return j.body, j.err
	case <-time.After(quick):
		return map[string]any{"state": "started", "ask": id,
			"note": "the print goes on; read its outcome with printer.print {ask}"}, nil
	case <-ctx.Done():
		return nil, fmt.Errorf("the caller went away; the print %s goes on: %w", id, ctx.Err())
	}
}

// PrintResult is the outcome of the print with this ask id: the device's answer when it is done (a failure is the
// error), and started, still printing, while it is not. A print this process followed is read from it, one it did not
// (it was restarted) from the device's answers, which are kept for 10 minutes.
func (s *Service) PrintResult(ctx context.Context, id string) (map[string]any, error) {
	if !askIDPattern.MatchString(id) {
		return nil, fmt.Errorf("ask %q is not an ask id: the 24 hex digits printer.print answered", id)
	}
	if j := s.Jobs.get(id); j != nil {
		return s.printWait(ctx, id, j)
	}
	raw, err := s.Asks.Answer(id)
	if err != nil {
		return nil, fmt.Errorf("the device process's answers cannot be read: %w", err)
	}
	if raw == nil {
		return map[string]any{"state": "started", "ask": id, "note": "still printing, or this ask is not known"}, nil
	}
	var out map[string]any
	if err := json.Unmarshal(raw, &out); err != nil {
		return nil, fmt.Errorf("the device process's answer is not JSON: %w", err)
	}
	if msg, bad := out["error"].(string); bad {
		return nil, fmt.Errorf("%s", msg)
	}
	return out, nil
}

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

// PrinterStatus is the printer's state over IPP and the scanner's over eSCL, none of it behind a login
// (ADR 0467 clause 5). A part that does not answer is an answer, with why, not an error. Which of printer and
// scanner is not set is said by name, and nothing is contacted.
func (s *Service) PrinterStatus(ctx context.Context) (map[string]any, error) {
	var unset []string
	if s.Settings.Printer == "" {
		unset = append(unset, "printer")
	}
	if s.Settings.Scanner == "" {
		unset = append(unset, "scanner")
	}
	if len(unset) > 0 {
		return nil, unsetError(unset...)
	}
	out := map[string]any{}
	if st, err := PrinterStateOf(ctx, s.Settings.Printer); err != nil {
		out["answers"] = false
		out["why"] = err.Error()
	} else {
		out["answers"] = true
		out["state"] = st.State
		out["state-code"] = st.StateCode
		out["state-reasons"] = st.Reasons
		out["make-and-model"] = st.MakeAndModel
		out["markers"] = st.Markers
	}
	if es, err := NewClient(s.Settings.Scanner).Status(ctx); err != nil {
		out["scanner"] = map[string]any{"answers": false, "why": err.Error()}
	} else {
		out["scanner"] = map[string]any{"answers": true, "state": es.State, "adf-state": es.AdfState}
	}
	return out, nil
}